package util;

import data.DataPoint;
import org.apache.beam.sdk.coders.AvroCoder;
import org.apache.beam.sdk.coders.DefaultCoder;
import org.apache.beam.sdk.transforms.Combine.CombineFn;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.util.ArrayList;

/**
 * Created by emily on 17-12-12.
 */
public class WindowMap extends CombineFn<DataPoint, WindowMap.Accum, WindowMap.Accum> {

    private static final Logger LOG = LoggerFactory.getLogger(WindowMap.class);

    @DefaultCoder(AvroCoder.class)
    public static class Accum implements Serializable{
        public static ArrayList<DataPoint> dpList = new ArrayList<DataPoint>();
    }


    @Override
    public Accum createAccumulator() { return new Accum(); }

    @Override
    public Accum addInput(Accum accum, DataPoint input) {
        LOG.info("addInput");
        accum.dpList.add(input);
        return accum;
    }

    @Override
    public Accum mergeAccumulators(Iterable<Accum> accums) {
        Accum merged = createAccumulator();
        for (Accum accum : accums) {
            for(DataPoint dp : accum.dpList){
                LOG.info("mergeAccumulators");
                merged.dpList.add(dp);
            }
        }
        return merged;
    }

    @Override
    public Accum extractOutput(Accum accum) {
        LOG.info("accum length:" + accum.dpList.size());
        return accum;
    }


}
